package com.zf.nospring.rocketmq.simple;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.junit.jupiter.api.Test;

import java.util.concurrent.TimeUnit;

/**
 * 发送消息
 */
public class ProducerTest {

    /**
     * 测试同步发送消息。发送方在发出消息后会在收到响应之后才发下一个数据包。一般用于重要通知消息，例如重要通知邮件、营销短信。
     */
    @Test
    public void testSendSync() throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        // 创建生产者，指定生产者所属的组
        DefaultMQProducer producer = new DefaultMQProducer("no-spring-pg");
        // 指定NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();
        for (int i = 1; i <= 5; i++) {
            // 创建消息对象：指定主题、标签、消息体
            Message message = new Message("test_topic", "myTag", ("同步测试消息" + i).getBytes());
            // 发送消息
            SendResult sendResult = producer.send(message);
            // 通过 sendResult.getSendStatus() 判断是否发送成功
            System.out.println(sendResult);
        }
        // 关闭生产者
        producer.shutdown();
    }

    /**
     * 异步发送消息，使用回调获取发送的结果，一般用于可能链路耗时较长而对响应时间敏感的业务场景
     */
    @Test
    public void testSendAsync() throws MQClientException, RemotingException, InterruptedException {
        // 创建生产者，指定生产者所属的组
        DefaultMQProducer producer = new DefaultMQProducer("no-spring-pg");
        // 指定NameServer的地址
        producer.setNamesrvAddr("localhost:9876");
        // 启动生产者
        producer.start();
        // 发送结果回调
        SendCallback sendCallback = new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("发送成功");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("发送失败");
            }
        };
        for (int i = 1; i <= 5; i++) {
            // 创建消息对象：指定主题、标签、消息体
            Message message = new Message("test_topic", "myTag", ("异步测试消息" + i).getBytes());
            // 异步发送消息
            producer.send(message, sendCallback);
        }

        // 等待 10s, 防止还没有发送，线程就结束了
        TimeUnit.SECONDS.sleep(10);
        // 关闭生产者
        producer.shutdown();
    }

}
